package momo.app.comms;

import java.io.IOException;
import java.net.ConnectException;
import java.net.InetAddress;
import java.util.Collection;
import java.util.Iterator;
import java.util.Map;
import java.util.Random;
import java.util.Set;
import java.util.StringTokenizer;
import java.util.concurrent.atomic.AtomicBoolean;

import momo.app.comms.monitor.TaskMonitor;
import momo.app.config.AppConfig;
import momo.app.multicast.Msg;
import momo.app.multicast.MsgRcvListener;
import momo.app.status.ElectionTimeout;
import momo.app.status.RootElection;
import momo.app.status.StatusManager;
import momo.app.structure.ChannelLink;
import momo.app.structure.Message;
import momo.app.structure.Peer;
import momo.app.structure.PeerLink;
import momo.app.structure.PendingLink;
import momo.app.structure.PendingLinkContainer;
import momo.app.structure.Relationship;
import momo.app.structure.RelationshipContainer;
import momo.app.util.ConnectionTest;
import momo.app.util.FileOutput;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;


public class CommsManager
{
	private Log log = LogFactory.getLog(CommsManager.class);
	private Log logStructure = LogFactory.getLog("MoMoStructureLog");
	private Log logSent = LogFactory.getLog("MoMoMsgSentLog");
	private Log logRcved = LogFactory.getLog("MoMoMsgRcvLog");
	
	private static CommsManager manager;
	private RcvListener listener;
	private RootElection elector;
	private ElectionTimeout electionTimeout;
	
	private PendingLinkContainer pendContainer;
	private RelationshipContainer rsContainer;
	private MsgRcvListener msgListener;
	
	private Boolean msgRcvedLock = new Boolean(false);
	private RcvLoopback loopback;
	
	private AtomicBoolean safeJoin = new AtomicBoolean(false);
	
	private TaskMonitor taskMonitor;
	
	private CommsManager()
	{
		pendContainer = new PendingLinkContainer();
		rsContainer = new RelationshipContainer();
		loopback = new RcvLoopback();
		taskMonitor = new TaskMonitor();
	}
	
	public static CommsManager getInstance()
	{
		if ( manager == null )
			manager = new CommsManager();
		return manager;
	}
	
	synchronized public void linkup(Peer parent, PeerLink link) throws IOException
	{
		SendChannel parentChannel = new SendChannel(parent.getAddressToUse(), parent.getCommPort());
		this.pendingPeer(parent, link, parentChannel);
		
		Peer me = AppConfig.getInstance().getMe();
		Message msg = new Message(Message.NO_FORWARD,
								me.getId(),
								Message.LINKUP,
								me.toString());
		
		parentChannel.put(msg.toString());
	}
	
	synchronized public void ackLinkup(RcvChannel rcvChannel, PendingLink pendLink) throws IOException
	{
		Peer child = pendLink.getPeer();
		PeerLink childLink = pendLink.getPeerLink();
		
		InetAddress address = child.getAddressToUse();
		if ( address == null )
		{
			ConnectionTest testConnection = new ConnectionTest(child);
			address = testConnection.testConnection();
			child.setAddressToUse(address);
		}
		
		if ( child.getAddressToUse() == null )
			throw new IOException();
		
		SendChannel childChannel = new SendChannel(child.getAddressToUse(), child.getCommPort());
		this.addChild(child, rcvChannel, childChannel);
		StatusManager statusManager = StatusManager.getInstance();
		statusManager.addPeer(child, "ackLinkup");
		statusManager.addPeerLink(childLink);
		
		Peer me = AppConfig.getInstance().getMe();
		
		childChannel.put(
				new Message( Message.NO_FORWARD,
						me.getId(), 
						Message.LINKUP_ACK,
						me.toString()
						).toString()
					);
		
		StatusManager.getInstance().broadcast(
					new Message( Message.BROADCAST,
						me.getId(), 
						Message.NEW_PEER_LINKUP,
						child.toString()+"|"+childLink.toString()
					));
		
		statusManager.calWorkload("ackLinkup");
		statusManager.noteTimeJoined(child.getId());
		statusManager.timeResetExistingPeers();
		statusManager.displayStabMap("ackLinkup");
		statusManager.addLatMeasureThread(child);
		
		//restructure when new join (for planetlab)
//		Delayer.getInstance().delayMsgSent();
//		this.startRestructure();
	}
	
	synchronized public void joinNewParent(Peer parent) throws IOException
	{
		if ( parent == null )
			throw new ConnectException("Parent is null");
		
		InetAddress address = parent.getAddressToUse();
		if ( address == null )
		{
			ConnectionTest testConnection = new ConnectionTest(parent);
			address = testConnection.testConnection();
			parent.setAddressToUse(address);
		}
		
		if ( parent.getAddressToUse() == null )
			throw new IOException();
		
		Peer me = AppConfig.getInstance().getMe();
		SendChannel parentChannel = new SendChannel(parent.getAddressToUse(), parent.getCommPort());
		
		if ( log.isDebugEnabled() )
			log.debug("Adding parent to sendChannel: " + parentChannel.toString());
		
		this.pendingPeer(parent, null, parentChannel);
		
		Message msg = new Message(Message.NO_FORWARD,
				me.getId(),
				Message.NEW_PARENT,
				me.toString());
		parentChannel.put(msg.toString());
	}
	
	synchronized public void ackJoinNewParent(RcvChannel rcvChannel, Peer child) throws IOException
	{
		Peer me = AppConfig.getInstance().getMe();
		
		InetAddress address = child.getAddressToUse();
		if ( address == null )
		{
			ConnectionTest testConnection = new ConnectionTest(child);
			address = testConnection.testConnection();
			child.setAddressToUse(address);
		}
		
		if ( child.getAddressToUse() == null )
			throw new IOException();
		
		SendChannel childChannel = new SendChannel(child.getAddressToUse(), child.getCommPort());
		
		childChannel.put(
				new Message( Message.NO_FORWARD,
						me.getId(), 
						Message.NEW_PARENT_ACK,
						me.toString()
						).toString()
					);
		
		if ( log.isDebugEnabled() )
			log.debug("Adding child to sendChannel: " + childChannel.toString());
		
		this.addChild(child, rcvChannel, childChannel);
	}
	
	public void sourceForward(Message msg, boolean toBeStamped)
	{
		Random rand = new Random();
		int seed = rand.nextInt();
		
		if (log.isDebugEnabled())
			log.debug("I wanna send this msg, MsgId: " +msg.getMsgId() +"\nContents: \n" + msg.toString()+"\n");
		if ( logSent.isDebugEnabled() )
			logSent.debug("I wanna send this msg, MsgId: " +msg.getMsgId() +"\nContents: \n" + msg.toString()+"\n");
		
		if (toBeStamped)
		{
			if ( log.isDebugEnabled() )
				log.debug("Msg Stamping in progress, MsgId: " + msg.getMsgId());
			
			CommsHistory.getInstance().stampOwnMessage(msg);
			
			if ( log.isDebugEnabled() )
				log.debug("Msg Stamped, MsgId: " + msg.getMsgId());
			
			if ( Delayer.getInstance().isDelayed() )
			{
				if ( log.isDebugEnabled() )
					log.debug("Msg delayed in progress, MsgId: "+msg.getMsgId());
				
				Delayer.getInstance().storeDelayedMsg(msg);
				
				if ( log.isDebugEnabled() )
					log.debug("Msg delayed, MsgId: "+msg.getMsgId());
			}else
			{
				if ( log.isDebugEnabled() )
					log.debug("System not delayed, proceed to send, MsgId: " + msg.getMsgId());
				
				send(msg);
			}
		}else
		{
			if ( log.isDebugEnabled() )
				log.debug("No stamping required, proceed to send, MsgId: " + msg.getMsgId());
			
			send(msg);
		}
		
		//cater to rcved message when app user send data 
		if ( msg.getMsgType().equals(Message.USER_DATA) )
		{
			if ( log.isDebugEnabled() )
				log.debug("Msg rcved is USER_DATA, on loopback now, MsgId: " + msg.getMsgId());
			if ( logSent.isDebugEnabled() )
				logSent.debug("Msg rcved is USER_DATA, loopbacking in progress, MsgId: " + msg.getMsgId());
			
			loopback.put(msg.toString());
			
			if ( log.isDebugEnabled() )
				log.debug("Msg had loopbacked, MsgId: " + msg.getMsgId());
			if ( logSent.isDebugEnabled() )
				logSent.debug("Msg had loopbacked, MsgId: " + msg.getMsgId());
		}
	}
	
	public void ping()
	{
		Peer me = AppConfig.getInstance().getMe();
		this.forwardUp(new Message( Message.FORWARD_UP,
				me.getId(), 
				Message.PING,
				me.toString()
				));
		safeJoin.set(false);
	}
	
	public void setSafeJoin()
	{
		safeJoin.compareAndSet(false, true);
	}
	
	public void resetSafeJoin()
	{
		safeJoin.compareAndSet(true, false);
	}
	
	public boolean getSafeJoin()
	{
		return safeJoin.get();
	}

	private void send(Message msg)
	{		
		if ( StatusManager.getInstance().isRoot() )
		{
			msg.setForwardType(Message.FORWARD_DOWN);
			forwardDown(msg);
		}else
		{
			msg.setForwardType(Message.FORWARD_UP);
			forwardUp(msg);
		}
	}
	
	public void forwardUp(Message msg)
	{
		if ( log.isDebugEnabled() )
			log.debug("before Forward Msg Up, MsgId: " + msg.getMsgId());
		if ( logSent.isDebugEnabled() )
			logSent.debug("before Forward Msg Up, MsgId: " + msg.getMsgId());
		
		SendChannel parentChannel = rsContainer.getParentChannel();
		if ( parentChannel != null )
		{
			parentChannel.put(msg.toString());
			
			if ( log.isDebugEnabled() )
				log.debug("Forward Msg Up, MsgId: " + msg.getMsgId() + "\n --> " + parentChannel.toString());
			if ( logSent.isDebugEnabled() )
				logSent.debug("Forward Msg Up, MsgId: " + msg.getMsgId() + "\n --> " + parentChannel.toString());
		}else
		{
			if ( log.isDebugEnabled() )
				log.debug("No parent to forward up, MsgId: " + msg.getMsgId());
			if ( logSent.isDebugEnabled() )
				logSent.debug("No parent to forward up, MsgId: " + msg.getMsgId());
		}
	}
	
	public void forwardDown(Message msg)
	{
		if ( log.isDebugEnabled() )
			log.debug("before forward down, msgid: "+msg.getMsgId());
		if ( logSent.isDebugEnabled() )
			logSent.debug("before forward down, msgid: "+msg.getMsgId());
		
		Collection<SendChannel> set = rsContainer.getChildrenChannel();
		if ( set.size() <= 0 )
		{
			if ( log.isDebugEnabled() )
				log.debug("No children to forward down, MsgId: " + msg.getMsgId());
			if ( logSent.isDebugEnabled() )
				logSent.debug("No children to forward down, MsgId: " + msg.getMsgId());
		}else
		{
			StringBuffer buffer = new StringBuffer("Forward Msg Down, MsgId: " + msg.getMsgId() );
			Iterator<SendChannel> iter = set.iterator();
			while (iter.hasNext())
			{
				SendChannel childChannel = iter.next();
				childChannel.put(msg.toString());
				buffer.append("\n --> " + childChannel.toString() );
			}
			if ( log.isDebugEnabled() )
				log.debug(buffer.toString());
			if ( logSent.isDebugEnabled() )
				logSent.debug(buffer.toString());
		}
	}
	
	synchronized public void pendingPeer(Peer peer, PeerLink link, SendChannel channel)
	{
		pendContainer.put(peer, link, channel);
	}
	
	synchronized public PendingLink removePendingPeer(String id)
	{
		return pendContainer.remove(id);
	}
	
	synchronized public boolean addParent(Peer peer, RcvChannel rC, SendChannel sC)
	{
		return rsContainer.addParent(peer, rC, sC);
	}
	
	synchronized public void addChild(Peer peer, RcvChannel rC, SendChannel sC)
	{
		rsContainer.addChildren(peer, rC, sC);
	}
	
	synchronized public Relationship remove(String id, RcvChannel rC)
	{
		return rsContainer.remove(id, rC);
	}
	
	synchronized public void listen()
	{
		if ( listener == null )
			listener = new RcvListener(AppConfig.getInstance().getCommPort());
	}
	
	synchronized public void stopListening()
	{
		if ( listener != null )
		{
			listener.stopListening();
			listener = null;
		}
	}
	
	synchronized public void stopAllComms()
	{
		this.stopListening();
		rsContainer.clear();
	}
	
	synchronized public boolean startElection(String electionType)
	{		
		if ( elector == null || !elector.isAlive() )
		{
			elector = new RootElection();
			elector.setElectionType(electionType);
			elector.start();
			startElectionTimeout();
			return true;
		}
		return false;
		
	}
	
	synchronized public boolean startElectionTimeout()
	{		
		if ( electionTimeout == null || !electionTimeout.isAlive() )
		{
			electionTimeout = new ElectionTimeout();
			electionTimeout.start();
			return true;
		}
		return false;
	}
	
	synchronized public void stopElection()
	{
		if ( elector != null )
		{
			if ( elector.isAlive() )
			{
				elector.stopEvent();
				if ( log.isDebugEnabled() )
					log.debug("Stop Election");
			}
		}
	}
	
	synchronized public void stopElectionTimeout()
	{
		if ( electionTimeout != null )
		{
			if ( electionTimeout.isAlive() )
			{
				electionTimeout.stopEvent();
				if ( log.isDebugEnabled() )
					log.debug("Stop Election Timeout");
			}
		}
	}
	
	synchronized public void vote(int candidateId)
	{
		if ( elector != null )
			elector.storeCandidateId(candidateId);
	}
	
	synchronized public void setMsgListener(MsgRcvListener msgListener)
	{
		this.msgListener = msgListener;
	}
	
	public void rcvMsg(Message message)
	{
		Random rand = new Random();
		int seed = rand.nextInt();
		
		if ( logRcved.isDebugEnabled() )
			logRcved.debug("Processing Msg to send to API, from: "+message.getSourceId()+" MsgId: " + message.getMsgId());
		if ( log.isDebugEnabled() )
			log.debug("Processing Msg to send to API, from "+message.getSourceId()+" MsgId: " + message.getMsgId());
		
		if ( log.isDebugEnabled() )
			log.debug("MsgListener is null? : " + (msgListener==null) );
		
		if ( msgListener != null )
		{
			String msgId = message.getMsgId();
			String source = message.getSourceId();
			String contents = message.getMsg();
			
			if ( log.isDebugEnabled() )			
				log.debug("Processing workload, MsgId: " + message.getMsgId() );
			
			StatusManager.getInstance().incrementWorkload(source);
			
			if ( log.isDebugEnabled() )
				log.debug("Workload Processed, MsgId: " + message.getMsgId() );
			
			Msg msg = new Msg(msgId, source, contents);

			if ( log.isDebugEnabled() )
				log.debug("Acquiring lock to send to API, MsgId: " + message.getMsgId());
			
			synchronized (msgRcvedLock)
			{
				if ( log.isDebugEnabled() )
					log.debug("Lock acquired, MsgId: " + message.getMsgId());
				
				taskMonitor.putTask(msgListener, msg);
				
				if ( logRcved.isDebugEnabled() )
					logRcved.debug("Msg send to API Listener, MsgId: " + message.getMsgId());
				if ( log.isDebugEnabled() )
					log.debug("Msg send to API Listener, MsgId: " + message.getMsgId());
			}
			
			if ( log.isDebugEnabled() )
				log.debug("Msg sent to API, MsgId: " + message.getMsgId());
		}
	}
	
	synchronized public void msgLost(Map<String, Set<Integer>> lostMsg)
	{
		if ( msgListener != null )
			msgListener.msgLost(lostMsg);
	}
	
	synchronized public void restructureRequired(Message msg)
	{
		if ( StatusManager.getInstance().isRoot() )
			startRestructure();
		else
		{
			if ( log.isDebugEnabled() )
				log.debug("who started my election timeout?");
			startElectionTimeout();
		}
	}
	
	synchronized public void startRestructure()
	{
		if ( elector == null )
			elector = new RootElection();
		
		if ( !elector.isAlive() )
			elector.restructure();
	}
	
	synchronized public void restructure(Message msg)
	{
		if ( log.isDebugEnabled() )
			log.debug("Restructure, MsgId: " + msg.getMsgId());
		if ( logStructure.isDebugEnabled() )
			logStructure.debug("Restructure, MsgId: " + msg.getMsgId());
		
		stopElectionTimeout();
		
		StatusManager statusManager = StatusManager.getInstance();
		Peer me = AppConfig.getInstance().getMe();
		
		StringTokenizer tokens = new StringTokenizer(msg.getMsg(), "|");
		double score = Double.valueOf( tokens.nextToken() );
		statusManager.setTreeScore(score);
		Peer root = Peer.createPeer(tokens.nextToken());
		
		statusManager.setRootId(root.getId());
		
		while ( tokens.hasMoreTokens() )
		{
			Peer child = Peer.createPeer(tokens.nextToken());
			Peer parent = Peer.createPeer(tokens.nextToken());
			
			String childId = child.getId();
			String parentId = parent.getId();
			
			if ( me.getId().equals( childId )) 
			{
//				not same parent?				
				if ( !rsContainer.isParent( parentId ) )
				{
					if ( log.isDebugEnabled() )
						log.debug("Diff parent as prev structure. Trying to join new parent");
					if ( logStructure.isDebugEnabled() )
						logStructure.debug("Diff parent as prev structure. Trying to join new parent");
					
					if ( rsContainer.isChild( parentId ) )
					{
						if ( log.isDebugEnabled() ) {
							log.debug("Special case: the new parent is currently my child");
							log.debug("Breaking the channel connection between this child");
						}
						
						if ( logStructure.isDebugEnabled() ) {
							logStructure.debug("Special case: the new parent is currently my child");
							logStructure.debug("Breaking the channel connection between this child");
						}
						
						try {
							Relationship rs = rsContainer.remove( parentId );
							rs.getChannels().initCloseConnection();
						}
						catch (IllegalArgumentException ie) {
							log.error(ie);
						}
					}
					
					if ( rsContainer.hasParent() )
					{
						if ( log.isDebugEnabled() )
							log.debug("Breaking the channel connection between my current parent");
						if ( logStructure.isDebugEnabled() )
							logStructure.debug("Breaking the channel connection between my current parent");
						
						Relationship rs = rsContainer.removeParent();
						rs.getChannels().initCloseConnection();
						
					}else
					{
						if ( log.isDebugEnabled() )
							log.debug("Parent not found. Channel was not broken for parent");
						if ( logStructure.isDebugEnabled() )
							logStructure.debug("Parent not found. Channel was not broken for parent");
					}
					
					try
					{
						this.joinNewParent(parent);
					}catch (ConnectException ce)
					{
						statusManager.removeAssociationOnPeer(parent.getId());
						if ( statusManager.getPeers().size() == 1 )
						{
							if ( log.isDebugEnabled() )
								log.debug("I m the only guy left in the network");
							if ( logStructure.isDebugEnabled() )
								logStructure.debug("I m the only guy left in the network");
							
							statusManager.setMeAsRoot();
							stopElectionTimeout();
						}else
						{
							if ( log.isDebugEnabled() )
								log.debug("Parent unable to join. Send restructure msg thro status channel");
							StatusManager.getInstance().broadcast(
									new Message(
											Message.BROADCAST, 
											parent.getId(), 
											Message.RESTRUCTURE_REQUIRED, 
											parent.getId()));
							
							if ( log.isDebugEnabled() )
								log.debug("who started my election timeout?");
							startElectionTimeout();
						}
					}catch (IOException ioe)
					{
						if ( log.isErrorEnabled() )
							log.error("IOException", ioe);
					}
					break;
				}else
				{
					if ( log.isDebugEnabled() )
						log.debug("Same parent as prev structure");
					if ( logStructure.isDebugEnabled() )
						logStructure.debug("Same parent as prev structure");
					
					break;
				}
			}
		}
		
		try {
			Thread.sleep(5 * 1000);
			
			if ( log.isDebugEnabled() )
				log.debug("I m freeing the delay - Restructure");
			
			Delayer.getInstance().freeDelay();
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
	}
	
	synchronized public void removeAssociationOnPeer(String id, RcvChannel rC)
	{
		if ( log.isDebugEnabled() )
			log.debug("removeAssociationOnPeer: " + id);
		if ( logStructure.isDebugEnabled() )
			logStructure.debug("removeAssociationOnPeer: " + id);
		
		StatusManager statusManager = StatusManager.getInstance();
		
		listener.removeDeadListener();
		
		rsContainer.display();
		
		Relationship rs = this.remove(id, rC);
		if ( rs != null )
		{
			rs.getChannels().closeChannels();
			
			if ( statusManager.getPeers().size() == 1 )
			{
				if ( log.isDebugEnabled() )
					log.debug("I m the only guy left in the network");
				if ( logStructure.isDebugEnabled() )
					logStructure.debug("I m the only guy left in the network");
				
				statusManager.setMeAsRoot();
				stopElectionTimeout();
				Delayer.getInstance().freeDelay();
			}else
			{
				if ( log.isDebugEnabled() )
					log.debug("Send restructure msg thro status channel");
				
				statusManager.broadcast(
						new Message(
								Message.BROADCAST, 
								AppConfig.getInstance().getId(), 
								Message.RESTRUCTURE_REQUIRED, 
								id));
				
				if ( log.isDebugEnabled() )
					log.debug("who started my election timeout?");
				
				startElectionTimeout();
				
				if ( !rs.isParent() && statusManager.isRoot() )				
					startRestructure();
			}
		}else
		{
			
		}
	}
	
	synchronized public void resetState()
	{
		StateResetter.getInstance().reset(0, true);
	}
	
}//end of class CommsManager_v2
